package com.baixiaowen.xiaoaointerview.并发提升篇.调度器and线程池;

import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class Scheduler {

    // LinkedTransferQueue 底层基于DualQueue和DualStack来实现反向传递压力的策略，即如果没有足够的消费者就等待或拒绝 是不公平的
    SynchronousQueue<Runnable> tasks = new SynchronousQueue<>(false);

    // LinkedTransferQueue 底层基于DualQueue来实现反向传递压力的策略，即如果没有足够的消费者就等待或拒绝 是公平的
//    LinkedTransferQueue<Runnable> tasks = new LinkedTransferQueue();

    static AtomicInteger idCount = new AtomicInteger(0);

    public Scheduler(int workers) {
        for (int i = 0; i < workers; i++) {
            new Thread(new Worker()).start();
        }
    }

    class Worker implements Runnable {

        int id;

        public Worker() {
            this.id = idCount.getAndIncrement();
        }

        @Override
        public void run() {
            while (true) {
                Runnable runnable = null;
                try {
                    runnable = tasks.take();
                    runnable.run();
                    System.err.format("work done by id=%d\n", id);
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public void submit(Runnable r) throws InterruptedException {
        // 如果调用 tasks.offer(r) 那么对于 LinkedTransferQueue 来说就是一个普通的队列，没有DualQueue的特点，就变成了LinkedBlockQueue
        tasks.offer(r);
//        while (!tasks.tryTransfer(r)) {
//            // Java 9 性能要比 Thread.sleep(0) 好
////            Thread.onSpinWait();
//            Thread.sleep(0);
//
//            new Thread(new Worker()).start();
//        }

        // SynchronizeQueue 中的 tasks.offer(r) 和 tasks.tryTransfer(r)的能力是近似的
        if (!tasks.offer(r)) {
            // Java 9 性能要比 Thread.sleep(0) 好
//            Thread.onSpinWait();
            Thread.sleep(0);
            new Thread(new Worker()).start();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Scheduler scheduler = new Scheduler(10);

        for (int i = 0; i < 1000; i++) {
            int localI = i;
            Thread.sleep(1);
            scheduler.submit(() -> {
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        
    }

}
